package net.wantongtech.xa.uploadbasicinfo.service;

import com.google.common.collect.Lists;
import com.jcraft.jsch.ChannelSftp;
import com.jcraft.jsch.SftpATTRS;
import com.jcraft.jsch.SftpException;
import lombok.extern.slf4j.Slf4j;
import net.wantongtech.xa.uploadbasicinfo.config.SSHServerConfig;
import net.wantongtech.xa.uploadbasicinfo.dao.ScanFastDFSDao;
import net.wantongtech.xa.uploadbasicinfo.util.SFTPHelper;
import net.wantongtech.xa.uploadbasicinfo.vo.DfsFileEntry;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.joda.time.LocalDate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;
import java.util.Vector;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * 扫描FastDFS存储服务实现
 *
 * @author MaShibo
 * 2021/1/12 0012
 */
@Slf4j
@Service
public class ScanFastDFSService {

    @Resource
    private SSHServerConfig sshServerConfig;

    @Resource
    private ScanFastDFSDao scanFastDFSDao;

    @Resource
    private ThreadPoolTaskExecutor taskExecutor;


    private AtomicInteger atomicFileId = new AtomicInteger(1);

    /**
     * 每次默认删除数量
     */
    public static final int DEFAULT_DEL_CNT = 5000;


    @SuppressWarnings("unchecked")
    public void scan() {
        try {
            SFTPHelper sftpHelper = getSftpHelperInstance();
            sftpHelper.connect();
            ChannelSftp sftp = sftpHelper.getSftp();

            String rootPath = sshServerConfig.getFastDfsBasePath();
            log.info("根扫描路径为：{}", rootPath);

            Vector<ChannelSftp.LsEntry> vector = (Vector<ChannelSftp.LsEntry>) sftp.ls(rootPath);


            List<ChannelSftp.LsEntry> lsEntryList = vector.stream()
                    .filter(lsEntry -> {
                        String filename = lsEntry.getFilename();
                        SftpATTRS attrs = lsEntry.getAttrs();
                        return !".".equals(filename) && !"..".equals(filename) && attrs.isDir();
                    }).collect(Collectors.toList());

            int total = lsEntryList.size();
            log.info("根路径下共有{}个文件夹需要扫描", total);


            int corePoolSize = taskExecutor.getCorePoolSize();
            log.info("线程池共有核心线程数：{}个", corePoolSize);

            int perNum = total / corePoolSize;
            log.info("每个线程扫描{}个文件夹，剩余{}个", perNum, total % corePoolSize);

            List<Future<Void>> futureList = Lists.newArrayList();

            for (int i = 1; i <= corePoolSize; i++) {
                int startIndex = (i - 1) * perNum;
                int endIndex = i * perNum;
                if (i == corePoolSize) {
                    endIndex = total;
                }
                List<ChannelSftp.LsEntry> subLsEntryList = lsEntryList.subList(startIndex, endIndex);
                List<String> dirPathList = subLsEntryList.stream()
                        .map(lsEntry -> String.format("%s/%s", rootPath, lsEntry.getFilename()))
                        .collect(Collectors.toList());
                ScanDirCallable scanDirCallable = new ScanDirCallable(startIndex + 1, endIndex, dirPathList);
                futureList.add(taskExecutor.submit(scanDirCallable));
            }

            for (Future<Void> future : futureList) {
                try {
                    future.get();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    log.error(e.getMessage(), e);
                } catch (ExecutionException e) {
                    log.error(e.getMessage(), e);
                }
            }

        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        log.info("文件扫描完成");
    }

    private SFTPHelper getSftpHelperInstance() {
        return new SFTPHelper(
                sshServerConfig.getHost(),
                sshServerConfig.getPort(),
                sshServerConfig.getUsername(),
                sshServerConfig.getPassword());
    }

    @SuppressWarnings("rawtypes")
    public void scanDir(String currentPath, ChannelSftp sftp) throws SftpException {
        Vector vector = sftp.ls(currentPath);
        for (Object obj : vector) {
            ChannelSftp.LsEntry lsEntry = (ChannelSftp.LsEntry) obj;
            String filename = lsEntry.getFilename();
            if (".".equals(filename) || "..".equals(filename)) {
                continue;
            }
            SftpATTRS attrs = lsEntry.getAttrs();
            String absolutePath = String.format("%s/%s", currentPath, filename);
            if (attrs.isDir()) {
                scanDir(absolutePath, sftp);
            }
            if (isRequiredFile(filename)) {
                long size = attrs.getSize();
                int mTime = attrs.getMTime();
                Date modifyDate = new Date(((long) mTime) * 1000L);
                String modifyDateStr = DateFormatUtils.ISO_8601_EXTENDED_DATETIME_FORMAT.format(modifyDate);
                log.info("文件名：{}，大小：{}，最后修改时间：{}", absolutePath, size, modifyDateStr);
                DfsFileEntry dfsFile = new DfsFileEntry();
                int fileId = atomicFileId.getAndIncrement();
                dfsFile.setId(fileId);
                dfsFile.setPath(absolutePath);
                dfsFile.setModifyTime(modifyDate);
                dfsFile.setFileSize(size);
                scanFastDFSDao.createValueObject(dfsFile);
            }
        }
    }

    private boolean isRequiredFile(String filename) {
        return filename.contains(".json") || StringUtils.endsWithIgnoreCase(filename, ".jpg") ||
                StringUtils.endsWithIgnoreCase(filename, ".png");
    }


    public void deleteFile(String maxModifyDateStr) {
        log.info("最大修改日期：{}", maxModifyDateStr);
        Date maxModifyDate = LocalDate.parse(maxModifyDateStr).toDate();
        int total = scanFastDFSDao.queryTotalCnt(maxModifyDate);
        log.info("最大修改日期前，共有{}个文件", total);
        int cnt = (int) Math.ceil((double) total / DEFAULT_DEL_CNT);
        log.info("每次删除{}个文件，共需要删除{}次", DEFAULT_DEL_CNT, cnt);
        List<Future<Void>> futureList = Lists.newArrayList();

        for (int i = 1; i <= cnt; i++) {
            int start = 1 + DEFAULT_DEL_CNT * (i - 1);
            int end = DEFAULT_DEL_CNT * i;
            if (end > total) {
                end = total;
            }
            Future<Void> future = taskExecutor.submit(new DeleteFileCallable(start, end, maxModifyDate));
            futureList.add(future);
        }

        for (Future<Void> future : futureList) {
            try {
                future.get();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                log.error(e.getMessage(), e);
            } catch (ExecutionException e) {
                log.error(e.getMessage(), e);
            }
        }
        log.info("删除完成");
    }


    public class ScanDirCallable implements Callable<Void> {

        private final int start;

        private final int end;

        private final List<String> dirPathList;

        public ScanDirCallable(int start, int end, List<String> dirPathList) {
            this.start = start;
            this.end = end;
            this.dirPathList = dirPathList;
        }

        @Override
        public Void call() throws Exception {
            log.info("开始扫描，start-{}，end-{}的文件", start, end);
            SFTPHelper sftpHelper = getSftpHelperInstance();
            sftpHelper.connect();
            ChannelSftp sftp = sftpHelper.getSftp();
            for (String dirPath : dirPathList) {
                scanDir(dirPath, sftp);
            }
            log.info("start-{}，end-{}的文件，扫描完成", start, end);
            try {
                return null;
            } finally {
                sftpHelper.disconnect();
            }
        }
    }


    public class DeleteFileCallable implements Callable<Void> {

        private final int start;
        private final int end;
        private final Date maxModifyDate;
        private SFTPHelper sftpHelper = getSftpHelperInstance();

        public DeleteFileCallable(int start, int end, Date maxModifyDate) {
            this.start = start;
            this.end = end;
            this.maxModifyDate = maxModifyDate;
        }

        @Override
        public Void call() {
            List<DfsFileEntry> dfsFileEntryList = scanFastDFSDao.queryDfsFileEntriesByRowAndModifyTime(start, end, maxModifyDate);
            int size = dfsFileEntryList.size();
            log.info("start：{}，end：{}，共查询出{}条数据", start, end, size);
            try {
                sftpHelper.connect();
                ChannelSftp sftp = sftpHelper.getSftp();
                try {
                    for (DfsFileEntry dfsFileEntry : dfsFileEntryList) {
                        String path = dfsFileEntry.getPath();
                        if (isRequiredFile(path)) {
                            try {
                                if (fileExists(path)) {
                                    sftp.rm(path);
                                    log.info("删除文件：{}", path);
                                } else {
                                    log.info("文件：{}不存在", path);
                                }
                                scanFastDFSDao.updateDeleteFlag(dfsFileEntry.getId());
                            } catch (Exception e) {
                                log.error(e.getMessage(), e);
                            }
                        }
                    }
                } finally {
                    sftpHelper.disconnect();
                }
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
            log.info("start：{}，end：{}，完成删除", start, end);
            return null;
        }

        private boolean fileExists(String filePath) throws SftpException {
            ChannelSftp sftp = sftpHelper.getSftp();
            try {
                sftp.lstat(filePath);
            } catch (SftpException e) {
                if (e.id == ChannelSftp.SSH_FX_NO_SUCH_FILE) {
                    return false;
                }
                throw e;
            }
            return true;
        }
    }

}
